package com.example.demo.service;

import com.alibaba.dashscope.audio.asr.recognition.Recognition;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionParam;
import com.alibaba.dashscope.audio.asr.recognition.RecognitionResult;
import com.alibaba.dashscope.exception.NoApiKeyException;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;

@Service
@Slf4j
public class SpeechRecognitionService {

    // 存储每个会话的音频数据队列
    private final ConcurrentHashMap<String, BlockingQueue<ByteBuffer>> audioQueues = new ConcurrentHashMap<>();
    
    // 存储每个会话的识别器
    private final ConcurrentHashMap<String, Recognition> recognizers = new ConcurrentHashMap<>();
    
    // 存储每个会话的状态
    private final ConcurrentHashMap<String, Boolean> sessionActive = new ConcurrentHashMap<>();

    /**
     * 开始语音识别会话
     * @param sessionId 会话ID
     * @param resultCallback 识别结果回调
     * @throws NoApiKeyException API密钥异常
     */
    public void startRecognition(String sessionId, Consumer<RecognitionResult> resultCallback) throws NoApiKeyException {
        log.info("=== 开始语音识别会话: {} ===", sessionId);
        
        // 创建音频数据队列
        BlockingQueue<ByteBuffer> audioQueue = new LinkedBlockingQueue<>();
        audioQueues.put(sessionId, audioQueue);
        sessionActive.put(sessionId, true);
        log.info("音频数据队列已创建: {}", sessionId);
        
        // 创建识别器
        Recognition recognizer = new Recognition();
        recognizers.put(sessionId, recognizer);
        log.info("语音识别器已创建: {}", sessionId);

        // 创建识别参数
        RecognitionParam param = RecognitionParam.builder()
                .model("paraformer-realtime-v2")
                .apiKey("sk-5597787e7bad498bb0a91054afc6c0f5")
                .format("pcm")
                .sampleRate(16000)
                .parameter("language_hints", new String[]{"zh", "en"})
                .build();
        
        log.info("识别参数已配置: 模型={}, 格式={}, 采样率={}", 
                param.getModel(), param.getFormat(), param.getSampleRate());
        
        // 创建Flowable，模仿测试代码的方式
        Flowable<ByteBuffer> audioSource = Flowable.create(emitter -> {
            new Thread(() -> {
                try {
                    log.info("音频流处理线程启动: {}", sessionId);
                    while (sessionActive.getOrDefault(sessionId, false)) {
                        try {
                            // 从队列中获取音频数据，超时1秒
                            ByteBuffer buffer = audioQueue.poll(1, java.util.concurrent.TimeUnit.SECONDS);
                            if (buffer != null) {
                                log.info("从队列获取音频数据: 会话={}, 数据长度={}字节", sessionId, buffer.remaining());
                                emitter.onNext(buffer);
                            }
                        } catch (InterruptedException e) {
                            log.warn("音频流处理被中断: {}", sessionId);
                            break;
                        }
                    }
                    log.info("音频流处理结束，发送完成信号: {}", sessionId);
                    emitter.onComplete();
                } catch (Exception e) {
                    log.error("音频流处理异常: 会话={}", sessionId, e);
                    emitter.onError(e);
                }
            }).start();
        }, BackpressureStrategy.BUFFER);
        
        log.info("音频源Flowable已创建: {}", sessionId);
        
        // 异步开始流式识别（避免阻塞WebSocket处理）
        new Thread(() -> {
            try {
                log.info("开始调用阿里云语音识别API: {}", sessionId);
                recognizer.streamCall(param, audioSource)
                        .blockingForEach(result -> {
                            String text = result.getSentence().getText();
                            boolean isFinal = result.isSentenceEnd();
                            log.info("收到识别结果: 会话={}, 文本='{}', 是否最终结果={}", 
                                    sessionId, text, isFinal);
                            resultCallback.accept(result);
                        });
                log.info("语音识别流处理完成: {}", sessionId);
            } catch (Exception e) {
                log.error("语音识别异常: 会话={}", sessionId, e);
            }
        }).start();
        
        log.info("语音识别线程已启动: {}", sessionId);
    }

    /**
     * 发送音频数据
     * @param sessionId 会话ID
     * @param audioData 音频数据
     */
    public void sendAudioData(String sessionId, byte[] audioData) {
        BlockingQueue<ByteBuffer> audioQueue = audioQueues.get(sessionId);
        if (audioQueue != null && sessionActive.getOrDefault(sessionId, false)) {
            // 创建新的ByteBuffer并设置limit，模仿测试代码的方式
            ByteBuffer buffer = ByteBuffer.allocate(audioData.length);
            buffer.put(audioData);
            buffer.flip(); // 设置为读模式
            
            try {
                audioQueue.offer(buffer);
                log.info("✓ 音频数据已加入队列: 会话={}, 数据长度={}字节, 队列大小={}", 
                        sessionId, audioData.length, audioQueue.size());
            } catch (Exception e) {
                log.error("✗ 音频数据加入队列失败: 会话={}", sessionId, e);
            }
        } else {
            log.warn("✗ 无法发送音频数据，会话不存在或已结束: {}", sessionId);
        }
    }

    /**
     * 结束语音识别会话
     * @param sessionId 会话ID
     */
    public void endRecognition(String sessionId) {
        log.info("结束语音识别会话: {}", sessionId);
        
        // 设置会话为非活跃状态
        sessionActive.put(sessionId, false);
        
        // 清理音频队列
        BlockingQueue<ByteBuffer> audioQueue = audioQueues.remove(sessionId);
        if (audioQueue != null) {
            audioQueue.clear();
            log.info("音频队列已清理: {}", sessionId);
        }
        
        // 关闭识别器
        Recognition recognizer = recognizers.remove(sessionId);
        if (recognizer != null) {
            try {
                recognizer.getDuplexApi().close(1000, "session_end");
                log.info("会话 {} 识别器已关闭", sessionId);
            } catch (Exception e) {
                log.error("关闭识别器异常: 会话={}", sessionId, e);
            }
        }
        
        // 清理会话状态
        sessionActive.remove(sessionId);
        log.info("会话状态已清理: {}", sessionId);
    }

    /**
     * 检查会话是否存在
     * @param sessionId 会话ID
     * @return 是否存在
     */
    public boolean isSessionActive(String sessionId) {
        return sessionActive.getOrDefault(sessionId, false);
    }
}